package com.datastream.process;

import entity.Topic1Data;
import init.InitConf;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import utils.JSONUtil;

import java.io.IOException;
import java.util.*;

/**
 * Created by kzdatd on 2020/1/13.
 */
public class MyDataProcesser implements Processor<String, String>, ProcessorSupplier<String, String> {

    private ProcessorContext processorContext;
    private static int i = 0;

    @Override
    public void init(ProcessorContext processorContext) {
        this.processorContext = processorContext;
    }

    @Override
    public void process(String key, String record) {
        {
            System.out.println("num :"+i);
            String val = record.toString();
            Map<String,Object> map = (Map<String, Object>) JSONUtil.JsonToObj(val,Map.class);
            i++;
            if (map == null){
                try {
                    throw new Exception("66666");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }else {
                List<Topic1Data> list = new ArrayList<>();
                map.forEach((v,k)->{
                    Map<String, Double> fields = new HashMap<>();
                    Map<String,String> tags = new HashMap<>();
                    fields.put("quality",192d);
                    fields.put("state",0d);
                    if (!v.equals("时间"))fields.put("value", (Double) k);
                    tags.put("node_id",InitConf.matchMap.get(v));
                    tags.put("device_id","");
                    Topic1Data topic1Data = new Topic1Data("data",new Date().getTime(),tags,fields);
                    if (!v.equals("时间"))list.add(topic1Data);
                });
                try {
                    record = JSONUtil.objectToStr(list);
                    processorContext.forward(key,record);
                } catch (IOException e) {
                    e.printStackTrace();
                    System.out.println("obj to json error ");
                }
            }}
    }

    @Override
    public void close() {

    }

    @Override
    public Processor get() {
        return new MyDataProcesser();
    }
}
